package org.pentaho.hadoop.shim.pvfs.conf;

import com.google.common.base.Preconditions;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.pentaho.di.connections.ConnectionDetails;

import java.net.URI;
import java.net.URISyntaxException;
import java.util.Arrays;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

import static com.google.common.base.Strings.isNullOrEmpty;
import static org.apache.hadoop.fs.Path.SEPARATOR;
import static org.pentaho.hadoop.shim.pvfs.PvfsHadoopBridge.getConnectionName;

public class OBSConf extends PvfsConf {

    private final String accessKey;
    private final String secretKey;
    private final String sessionToken;
    private String endpoint;
    private String pathStyleAccess;
    private final String credentialsFilePath;

    public OBSConf( ConnectionDetails details ) {
        super( details );
        Map<String, String> props = details.getProperties();
        credentialsFilePath = getVar( props, "credentialsFilePath" );

        accessKey = getVar( props, "accessKey" );
        secretKey = getVar( props, "secretKey" );
        sessionToken = getVar( props, "sessionToken" );
        // Use only when VFS is configured for generic OBS connection
        endpoint = getVar( props, "endpoint" );
        pathStyleAccess = getVar( props, "pathStyleAccess" );

    }

    @Override public boolean supportsConnection() {
        return Arrays.asList( "obs" ).contains( details.getType() );
    }

    @Override public Path mapPath(Path pvfsPath ) {
        validatePath( pvfsPath );
        String[] splitPath = pvfsPath.toUri().getPath().split( "/" );

        Preconditions.checkArgument( splitPath.length > 0 );
        String bucket = splitPath[ 1 ];
        String path = SEPARATOR + Arrays.stream( splitPath ).skip( 2 ).collect( Collectors.joining( SEPARATOR ) );
        try {
            return new Path( new URI( "obs", bucket, path, null ) );
        } catch ( URISyntaxException e ) {
            throw new IllegalStateException( e );
        }
    }

    @Override public Path mapPath( Path pvfsPath, Path realFsPath ) {
        URI uri = realFsPath.toUri();
        return new Path( pvfsPath.toUri().getScheme(),
                getConnectionName( pvfsPath ), "/" + uri.getHost() + uri.getPath() );
    }

    @Override public Configuration conf(Path pvfsPath ) {
        validatePath( pvfsPath );
        Configuration conf = new Configuration();
        conf.set( "fs.obs.access.key", accessKey );
        conf.set( "fs.obs.secret.key", secretKey );
        conf.set("fs.AbstractFileSystem.obs.impl", "org.apache.hadoop.fs.obs.OBS");
        conf.set( "fs.obs.impl", "org.apache.hadoop.fs.obs.OBSFileSystem" );
        conf.set( "fs.obs.connection.ssl.enabled", "true" );
        conf.set( "fs.obs.buffer.dir", System.getProperty( "java.io.tmpdir" ) );

        // Use only when VFS is configured for generic OBS connection
        if ( !isNullOrEmpty( endpoint ) ) {
            conf.set( "fs.obs.endpoint", "obs." + endpoint + ".myhuaweicloud.com" );
        }

        return conf;
    }

    @Override public boolean equals( Object o ) {
        if ( this == o ) {
            return true;
        }
        if ( o == null || getClass() != o.getClass() ) {
            return false;
        }
        if ( !super.equals( o ) ) {
            return false;
        }
        OBSConf obsConf = (OBSConf) o;
        return Objects.equals( accessKey, obsConf.accessKey )
                && Objects.equals( secretKey, obsConf.secretKey )
                && Objects.equals( sessionToken, obsConf.sessionToken )
                && Objects.equals( endpoint, obsConf.endpoint )
                && Objects.equals( pathStyleAccess, obsConf.pathStyleAccess )
                && Objects.equals( credentialsFilePath, obsConf.credentialsFilePath );
    }

    @Override public int hashCode() {
        return Objects.hash( super.hashCode(), accessKey, secretKey, sessionToken, endpoint, pathStyleAccess,
                credentialsFilePath );
    }
}
